package cn.tedu.charging.order.consumer;

import cn.tedu.charging.common.pojo.entity.ProgressEntity;
import cn.tedu.charging.common.pojo.message.ChargingCheckResultMessage;
import cn.tedu.charging.common.pojo.message.ChargingProgressMsg;
import cn.tedu.charging.common.pojo.message.ProgressData;
import cn.tedu.charging.common.pojo.param.ProgressCostParam;
import cn.tedu.charging.common.pojo.po.ChargingBillFailPO;
import cn.tedu.charging.common.pojo.po.ChargingBillSuccessPO;
import cn.tedu.charging.common.pojo.vo.ProgressCostVO;
import cn.tedu.charging.common.protocol.JsonResult;
import cn.tedu.charging.common.protocol.WebSocketMessage;
import cn.tedu.charging.order.context.MqttContext;
import cn.tedu.charging.order.dao.mapper.ChargingBillSuccessMapper;
import cn.tedu.charging.order.dao.repository.ProgressESRepository;
import cn.tedu.charging.order.feign.CostClient;
import cn.tedu.charging.order.webSocket.WebSocketServer;
import com.alibaba.fastjson2.JSON;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.BeanUtils;

import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Date;
import java.util.concurrent.TimeUnit;

@Slf4j
public class MqttConsumer implements MqttCallbackExtended {
    private MqttClient mqttClient;
    private MqttContext mqttContext;

    public MqttConsumer(MqttClient mqttClient, MqttContext mqttContext) {
        this.mqttClient = mqttClient;
        this.mqttContext = mqttContext;
    }

    /**
     * 连接成功 客户端mqttClient成功的建立了和emqx连接
     *
     * @param reconnect mqttClient 回调消费传递的参数 告知 方法 本次连接建立是否支持重连
     * @param serverURI 服务端emqx地址
     */
    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        try {
            mqttClient.subscribe("/charging/check/result");
            mqttClient.subscribe("/charging/progress");
        } catch (MqttException e) {
            log.error("订阅主题失败", e);
        }
    }

    /**
     * 连接断开 mqttClient断开和服务连接 回调的方法
     * 消费者使用的一些资源 消耗程序内容 在连接断开时候 释放避免因为这个消费者无法使用消费逻辑导致程序阻塞
     *
     * @param cause the reason behind the loss of connection.
     */
    @Override
    public void connectionLost(Throwable cause) {
        log.error("连接断开", cause);
    }

    /**
     * 消费者通过mqttClient监听订阅的主题 产生了消息,消息到达了当前消费对象
     *
     * @param topic   消息主题来源 主题名称
     * @param message 消息对象
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        log.debug("主题:{},消息:{}", topic, message);
        if (topic != null && topic.equals("/charging/check/result")) {
            handleCheckResult(topic, message);
        } else if (topic != null && topic.equals("/charging/progress")) {
            handleChargingProgress(topic, message);
        }
    }

    private void handleChargingProgress(String topic, MqttMessage message) {
        log.debug("当前消费者,主题:{},获取到消息:{},开始执行充电进度消费逻辑", topic, message);
        //1. 解析消息对象，转化为message数据封装类
        byte[] payload = message.getPayload();
        String jsonMsg = new String(payload, StandardCharsets.UTF_8);
        ChargingProgressMsg progress = JSON.parseObject(jsonMsg, ChargingProgressMsg.class);
        //2. UNDO 根据消息中传递的温度数据，判断是否安全，不安全就停止充电，通知手机操作
        checkTemperature(progress);
        //3. 远程调用计价中心，计算出本次充电的消费金额，返回ProgressVO计价vo对象完成后续流程
        ProgressCostVO vo = calculate(progress);
        //4. 封装一个ES的实体，记录充电进度信息，使用es客户端Repository，写入es存储
        saveProgress(progress,vo.getTotalCost());
        //5. 判断本次充电是否完成
        Boolean isFull = progress.getIsFull();
        if (isFull) {
            //状态修改
            ChargingBillSuccessMapper chargingBillSuccessMapper = mqttContext.getChargingBillSuccessMapper();
            UpdateWrapper wrapper=new UpdateWrapper();
            wrapper.eq("bill_id",progress.getOrderNo());
            wrapper.setSql("bill_status=2");
            chargingBillSuccessMapper.update(null,wrapper);
            //通知用户 充电最终金额 totalCost 充电结束 webSocketMessage.state=1 前端不展示
            WebSocketServer webSocketServer = mqttContext.getWebSocketServer();
            WebSocketMessage<String> webSocketMessage= new WebSocketMessage<>();
            webSocketMessage.setState(1);
            webSocketMessage.setMessage("over");
            webSocketMessage.setData("A 本次充电已完成，次消费金额："+vo.getTotalCost());
            webSocketServer.pushMessage(progress.getUserId(), JSON.toJSONString(webSocketMessage));
        } else {
            WebSocketServer webSocketServer = mqttContext.getWebSocketServer();
            WebSocketMessage<ProgressData> webSocketMessage= new WebSocketMessage<>();
            ProgressData data = new ProgressData();
            data.setChargingCapacity(progress.getChargingCapacity());
            data.setTotalCapacity(100.00);
            data.setOneElectricityCost(vo.getPowerFee());
            data.setTotalCost(vo.getTotalCost());
            data.setHours(0);
            data.setMinutes(10);
            data.setSeconds(10);
            webSocketMessage.setState(3);
            webSocketMessage.setMessage("charging");
            webSocketMessage.setData(data);
            webSocketServer.pushMessage(progress.getUserId(), JSON.toJSONString(webSocketMessage));
        }
    }

    private  void saveProgress(ChargingProgressMsg progress, Double totalCost) {
        ProgressESRepository progressESRepository = mqttContext.getProgressESRepository();
        ProgressEntity entity = new ProgressEntity();
        BeanUtils.copyProperties(progress, entity);
        entity.setTotalCost(totalCost);
        String id = progress.getOrderNo();
        entity.setId(id);
        progressESRepository.save(entity);
    }

    private ProgressCostVO calculate(ChargingProgressMsg progress) {
        log.debug("订单解析消息,调用计价中心,计算价格,本次订单:{}", progress.getOrderNo());
        CostClient costClient = mqttContext.getCostClient();
        ProgressCostParam param = new ProgressCostParam();
        BeanUtils.copyProperties(progress, param);
        JsonResult<ProgressCostVO> result = costClient.calculate(param);
        log.debug("远程调用计价中心,计算价格,返回结果:{}", result);
        return result.getData();
    }

    private void checkTemperature(ChargingProgressMsg progress) {
    }

    private void handleCheckResult(String topic, MqttMessage message) {
        log.debug("当前消费者,主题:{},获取到消息:{},开始处理设备自检反馈", topic, message);
        log.debug("添加抢锁流程");
        String lockKey = "charging:order:" + Arrays.toString(message.getPayload());
        Boolean res = mqttContext.getRedisTemplate().opsForValue().setIfAbsent(lockKey, "", 5, TimeUnit.SECONDS);
        if (Boolean.FALSE.equals(res)) {
            log.debug("抢锁失败");
            return;
        }
        String msgJson = new String(message.getPayload(), StandardCharsets.UTF_8);
        ChargingCheckResultMessage msg = JSON.parseObject(msgJson, ChargingCheckResultMessage.class);
        Boolean result = msg.getResult();
        if (result) {
            ChargingBillSuccessPO po = new ChargingBillSuccessPO();
            po.setBillId(msg.getOrderNo());
            po.setUserId(msg.getUserId());
            po.setBillStatus(1);
            po.setCreateTime(new Date());
            mqttContext.getChargingBillSuccessMapper().insert(po);
        } else {
            ChargingBillFailPO po = new ChargingBillFailPO();
            po.setBillId(msg.getOrderNo());
            po.setUserId(msg.getUserId());
            po.setCreateTime(new Date());
            po.setFailDesc("设备自检反馈失败");
            mqttContext.getChargingBillFailMapper().insert(po);
            WebSocketMessage<String> webSocketMessage = new WebSocketMessage<>();
            webSocketMessage.setState(1);
            webSocketMessage.setMessage("fail");
            webSocketMessage.setData("B 设备有故障，请换一把枪");
            mqttContext.getWebSocketServer().pushMessage(msg.getUserId(), JSON.toJSONString(webSocketMessage));
        }
    }

    /**
     * 消息投递 消息消费彻底完成
     *
     * @param token message身份标识
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        log.debug("消息投递完成");
    }
}
